﻿#include "../../inc/helper.hpp"

const static int g_heart_beat_liveness = 3;
const static int g_heart_beat_interval = 1000;
const static int g_interval_init = 1000;
const static int g_interval_max = 32000;

const static char* g_msg_worker_ready = "\001";
const static char* g_msg_heart_beat = "\002";

zmq::socket_t worker_new(zmq::context_t& ctx)
{
	zmq::socket_t worker(ctx, zmq::socket_type::dealer);
	worker.connect("tcp://localhost:5556");

	info("worker ready");
	send_string(worker, g_msg_worker_ready);
	return worker;
}

int main()
{
	zmq::context_t ctx;
	zmq::socket_t worker = worker_new(ctx);

	int liveness = g_heart_beat_liveness;
	int interval = g_interval_max;

	Time heartbeat_at = Now() + absl::Milliseconds(g_heart_beat_interval);

	int cycles = 0;
	while (1)
	{
		zmq::pollitem_t item = { worker, 0, ZMQ_POLLIN, 0 };
		zmq::poll(&item, 1, g_heart_beat_interval);

		if (item.revents & ZMQ_POLLIN)
		{
			auto msgs = zmsg_recv(worker);
			if (msgs.size() == 3)
			{
				++cycles;

				if (cycles > 3 && rand() % 5 == 0)
				{
					info("simulating a crash");
					break;
				}
				else if (cycles > 3 && rand() % 5 == 0)
				{
					info("simulating CPU overload");
					std::this_thread::sleep_for(std::chrono::seconds(3));
				}

				info("normal reply");
				std::this_thread::sleep_for(std::chrono::seconds(1));
				zmsg_send(worker, msgs);
				liveness = g_heart_beat_liveness;
			}
			else if (msgs.size() == 1)
			{
				if (to_string(msgs[0]) == g_msg_heart_beat)
				{
					liveness = g_heart_beat_liveness;
				}
				else
				{
					info("invalid message");
					dump_msgs(msgs);
				}
			}
			else
			{
				info("invalid message");
				dump_msgs(msgs);
			}
			interval = g_interval_init;
		}
		else if( --liveness == 0 )
		{
			info("heartbeat failure, can't reach queue");
			info("reconnecting in {} msec", interval);
			std::this_thread::sleep_for(std::chrono::milliseconds(interval));

			if (interval < g_interval_max)
			{
				interval *= 2;
			}

			worker = worker_new(ctx);
			liveness = g_heart_beat_liveness;
		}
		
		if (Now() > heartbeat_at)
		{
			heartbeat_at = Now() + absl::Milliseconds(g_heart_beat_interval);
			send_string(worker, g_msg_heart_beat);
		}

	}
}